package com.ysy.mq.db;


import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.google.common.util.concurrent.RateLimiter;
import com.ysy.constant.RocketMqConstant;
import com.ysy.mapper.UserCountMapper;
import com.ysy.persimmons.common.util.StringUtils;
import jakarta.annotation.Resource;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@RocketMQMessageListener(
        consumerGroup = "persimmons_count_group"+RocketMqConstant.FANS_COUNT_DB_TOPIC,
        topic = RocketMqConstant.FANS_COUNT_DB_TOPIC,messageModel = MessageModel.CLUSTERING)
public class FansCountDBConsumer implements RocketMQListener<String> {

    private static Logger log = LoggerFactory.getLogger(FansCountDBConsumer.class);


    @Resource
    private UserCountMapper userCountMapper;

    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Override
    public void onMessage(String message) {
        // 判断类型是关注还是取消
        if (StringUtils.isBlank(message)) {
            log.warn("当前mq消费数据库为空，暂不处理");
            return;
        }

        rateLimiter.acquire();

        Map<Long,Integer> mapCount = JSON.parseObject(message,new TypeReference<Map<Long, Integer>>(){});

        if (CollUtil.isEmpty(mapCount)) {
            return;
        }
        // mybatis plus自带的没有这样写的效率高，plus需要查询2次
        mapCount.forEach((k,v)->{userCountMapper.updateOrInsertFansCount(k,v);});
    }
}
